package com.intct.dws;

import com.intct.func.AdsLocusMapFunc;
import com.intct.func.HBaseSinkFunc;
import com.intct.func.RedisSinkFunc;
import com.intct.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-09-22 09:44
 *
 * sink : redis hbase
 *      // {"driver_id":"11","lng":"104.05166","create_time":"2019-04-19 13:55:37","driver_type":"1","mobile":"1861***1024","driver_name":"邵5bfb8fd1c7af69efee217c72fabec38b","cancel_count":"0","oder_id_desen":"175********682","register_city":"110108","time":"1475976832","id":"11","order_id":"17592719043682","lat":"30.70214","driver_management_id":"0"}
 *      redis条件：
            key_name = order_id
            分数：time
            value: lng + ":" + lat
 *
 *      hbase条件：
 *          // {"rowkey":"511603","sink_family":"info","after":{"code":"511603","name":"四川省广安市前锋区"},"table":"dim_area_info"}
 *          // {"driver_id":"11","lng":"104.05166","create_time":"2019-04-19 13:55:37","driver_type":"1","mobile":"1861***1024","driver_name":"邵5bfb8fd1c7af69efee217c72fabec38b","cancel_count":"0","oder_id_desen":"175********682","register_city":"110108","time":"1475976832","id":"11","order_id":"17592719043682","lat":"30.70214","driver_management_id":"0"}
 *             family : info
 *             table  : ads_locus
 *             rowkey : order_id + driver_id + time
 *
 *             travel_config.ads_table_config
 *
 *
 */
public class DwsLocusAPI {
    public static void main(String[] args) throws Exception {
        // 配置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 从工程中resoucres路径读取配置文件
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(DwsLocusAPI.class.getResource("/intct.properties").getFile());

        // 数据源配置 ods kafka
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(parameterTool, "dwd_locus");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dwd_locus_source");

        // Sink到Redis
        sourceDS.addSink(new RedisSinkFunc());

        // Sink到HBase
        // 转换
        byte[][] splitKeys = new byte[7][];
        splitKeys[0] = "0001".getBytes();
        splitKeys[1] = "0002".getBytes();
        splitKeys[2] = "0003".getBytes();
        splitKeys[3] = "0004".getBytes();
        splitKeys[4] = "0005".getBytes();
        splitKeys[5] = "0006".getBytes();
        splitKeys[6] = "0007".getBytes();

        sourceDS.map(new AdsLocusMapFunc()).addSink(new HBaseSinkFunc("intct84", splitKeys));

        // 启动作业
        env.execute("ads_locus_jobs");
    }
}
